package com.github.dote.mio.client;

import com.alibaba.fastjson.JSONObject;
import com.github.dote.mio.ack.AckReceiveTask;
import com.github.dote.mio.ack.AckSendTask;
import com.github.dote.mio.config.IOClientConfig;
import com.github.dote.mio.daemon.KeepAliveDaemon;
import com.github.dote.mio.event.ConnectionListener;
import com.github.dote.mio.event.EventListener;
import com.github.dote.mio.event.EventManager;
import com.github.dote.mio.protocal.Packet;
import com.github.dote.mio.protocal.PacketType;
import com.github.dote.mio.protocal.VoidData;
import com.github.dote.mio.rtt.RTTHandler;
import com.github.dote.mio.util.CommonUtil;
import com.github.dote.mio.util.JsonUtil;
import com.github.dote.mio.util.ReflectionUtil;
import lombok.Data;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import sun.nio.ch.DirectBuffer;

import java.io.IOException;
import java.net.*;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Iterator;
import java.util.concurrent.*;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
@Getter
public class IOClient {

    private Executor executor = Executors.newSingleThreadExecutor();
    private Executor executor1 = Executors.newSingleThreadExecutor();
    private DatagramChannel channel = null;
    private Selector selector = null;
    private InetSocketAddress serverAddr;
    private InetSocketAddress localAddr;
    private volatile boolean connected = false;
    private boolean isFirst = true;
    private boolean canReconnect = false;

    private IOClientConfig config;
    private PacketHandler packetHandler;
    private KeepAliveDaemon keepAliveDaemon;
    private AckSendTask ackSendTask;
    private AckReceiveTask ackReceiveTask;
    private EventManager eventManager;
    private RTTHandler rttHandler;

    private final Object sendReceiveLock = new Object();
    private final Semaphore sendReceiveAutoEvent = new Semaphore(1);

    public IOClient(IOClientConfig config) {
        this.config = config;
        packetHandler = new PacketHandler(this);

        serverAddr = new InetSocketAddress(getConfig().getServerIp(), getConfig().getServerPort());
        localAddr = new InetSocketAddress(getConfig().getLocalPort());

        keepAliveDaemon = new KeepAliveDaemon(this);

        KeepAliveDaemon.KeepAliveCallback callback = new KeepAliveDaemon.KeepAliveCallback() {
            @Override
            public void onLostKeepAlive() {
                setConnected(false);
                reconnect();
            }
        };
        keepAliveDaemon.setCallback(callback);
        ackSendTask = new AckSendTask(this);
        ackReceiveTask = new AckReceiveTask(this);
        eventManager = new EventManager();
        rttHandler = new RTTHandler(config.getAckTimeout());
    }

    public void connect() {
        try {
            if (channel != null)
                innerClose();
            if (selector != null)
                selector.close();
            canReconnect = true;
            this.channel = DatagramChannel.open();
            selector = Selector.open();
            channel.configureBlocking(false);

            keepAliveDaemon.start(false);
            ackSendTask.startup(true);
            ackReceiveTask.startup(true);

            if (channel == null || selector == null) return;
            executor1.execute(() -> {
                try {
                    channel.configureBlocking(false);
                    channel.connect(serverAddr);
                    channel.register(selector, SelectionKey.OP_READ);

                    while (selector.select() > 0) {
                        Iterator iterator = selector.selectedKeys().iterator();
                        while (iterator.hasNext()) {
                            SelectionKey key = null;
                            try {
                                key = (SelectionKey) iterator.next();
                                iterator.remove();
                                if (key.isReadable()) {
                                    reveice(key);
                                }
                                if (key.isWritable()) {
                                }
                            } catch (IOException e) {
                                e.printStackTrace();
                                try {
                                    if (key != null) {
                                        key.cancel();
                                        key.channel().close();
                                    }
                                } catch (IOException cex) {
                                    e.printStackTrace();
                                }
                            }
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }

        keepAliveDaemon.send();

    }

    synchronized public void reveice(SelectionKey key) throws IOException {
        String threadName = Thread.currentThread().getName();
        if (key == null)
            return;
        try {
            DatagramChannel sc = (DatagramChannel) key.channel();
            ByteBuffer buf = ByteBuffer.allocate(4096);
            buf.clear();
            SocketAddress address = sc.receive(buf);
            buf.flip();
            setConnected(true);
            while (buf.hasRemaining()) {
                buf.get(new byte[buf.limit()]);
                byte[] buff = buf.array();
                packetHandler.messageReceived(buff, buff.length);
            }
            buf.clear(); // make buffer ready for writing次

        } catch (PortUnreachableException ex) {
            setConnected(false);
        }
    }


    private void reconnect() {
        if (!canReconnect)
            return;
        if (getConfig().isAutoReConnect()) {
            if (getConfig().isDebug()) {
                log.debug("Reconnecting..............");
            }
            try {
                keepAliveDaemon.start(false);
            } catch (Exception e) {

            }
        }
    }

    private void innerClose() throws Exception {
        if (keepAliveDaemon != null)
            keepAliveDaemon.stop();
        if (ackSendTask != null)
            ackSendTask.stop();
        if (ackReceiveTask != null)
            ackReceiveTask.stop();
        if (channel != null && channel.isConnected()) {
            channel.close();
        }
        channel = null;
        if (selector != null)
            selector.close();
        selector = null;
        setConnected(false);
    }

    public void close() {
        sendToServer("exit",new VoidData());
        try {
            innerClose();
        } catch (Exception e) {
            e.printStackTrace();
        }
        canReconnect = false;
    }

    public synchronized boolean send(Packet packet) {
        if (channel != null && packet != null && (packet.getType() != PacketType.EVENT_MESSAGE || isConnected())) {
            try {
                byte[] buff = packet.toBytes();
                ByteBuffer buf = ByteBuffer.allocate(buff.length);
                buf.clear();
                buf.put(buff);
                buf.flip();
                if (channel.write(buf) > 0) {
                    channel.register(selector, SelectionKey.OP_READ);
                    if (packet.isAck() && packet.getAckId() != null && !"".equals(packet.getAckId()) && packet.getType() == PacketType.EVENT_MESSAGE) {
                        if (!ackSendTask.exist(packet.getAckId())) {
                            ackSendTask.put(IOClient.this, packet);
                        }
                    }
                    buf.clear();
                    if (buf.isDirect()) {
                        ((DirectBuffer) buf).cleaner().clean();
                    }
                    return true;
                }
                buf.clear();
                if (buf.isDirect()) {
                    ((DirectBuffer) buf).cleaner().clean();
                }


            } catch (Exception e) {
            }
        }
        setConnected(false);
        return false;
    }

    public boolean sendToClient(String sessionId, String eventName, Object data, boolean ack) {
        Packet packet = new Packet();
        packet.setType(PacketType.EVENT_MESSAGE);
        packet.setSubType(eventName);
        packet.setDataContent(JsonUtil.toJsonString(data));
        packet.setSessionId(sessionId);
        packet.setAck(ack);
        if (ack)
            packet.setAckId(CommonUtil.createUniqueId());
        return send(packet);
    }

    public boolean sendToClient(String sessionId, String eventName, Object data) {
        return sendToClient(sessionId, eventName, data, true);
    }

    public boolean sendToServer(String eventName, Object data, boolean ack) {
        return sendToClient("0", eventName, data, ack);
    }

    public boolean sendToServer(String eventName, Object data) {
        return sendToClient("0", eventName, data, true);
    }


    public <T> T sendToClientReceive(String sessionId, String eventName, Object data, String receiveEventName, long sendTimeout,Class<T> clazz) {
        final Lock lock=new ReentrantLock();
        final Condition condition =lock.newCondition();
        final SendReceiveResult<T> result = new SendReceiveResult<>();
        IOClient.this.eventManager.addEventListener(receiveEventName, new EventListener<T>() {
            @Override
            public void onEvent(IOClient client, T data) {

                try{
                    lock.lock();
                    if(data instanceof  JSONObject){
                        JSONObject jsonObj= (JSONObject) data;
                        result.setData(jsonObj.toJavaObject(clazz));
                    }else{
                        result.setData(data);
                    }
                    condition.signal();
                }finally {
                    lock.unlock();
                }
            }
        });
        sendToClient(sessionId,eventName,data);
        try{
            lock.lock();
            condition.await(sendTimeout, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }

        return result.getData();
    }

    public <T> T sendToClientReceive(String sessionId, String eventName, Object data, String receiveEventName,Class<T> clazz) {
        return sendToClientReceive(sessionId, eventName, data, receiveEventName, 5000l,clazz);
    }

    public <T> T sendToServerReceive(String eventName, Object data, String receiveEventName, long sendTimeout,Class<T> clazz) {
        return sendToClientReceive("0", eventName, data, receiveEventName, sendTimeout,clazz);
    }

    public <T> T sendToServerReceive(String eventName, Object data, String receiveEventName,Class<T> clazz) {
        return sendToClientReceive("0", eventName, data, receiveEventName, 5000l,clazz);
    }

    private void setConnected(boolean value) {
        if (connected != value) {
            connected = value;

            if (value) {

                if (isFirst) {
                    if (getConfig() != null && getConfig().isDebug() == true) {
                        log.debug("Successfually connect");
                    }
                    executor.execute(() -> {
                        eventManager.onConnect(this);
                    });

                } else {
                    if (getConfig() != null && getConfig().isDebug() == true) {
                        log.debug("Successfually reconnect");
                    }
                    executor.execute(() -> {
                        eventManager.onReconnect(this);
                    });

                }
                if (isFirst) {
                    isFirst = false;
                }

            } else {
                if (getConfig() != null && getConfig().isDebug() == true) {
                    log.debug("Disconnect");
                }
                executor.execute(() -> {
                    eventManager.onDisconnect(this);
                });

            }
        }
    }
}
